#include "node_environment/node_environment.h"
#include "utils/logger.h"
#include "common.h"
#include "Network.h"


//const int metaDataOffset = sizeof(uint32_t) + 1;
const int metaFlagOffset = sizeof(uint32_t);
const int metaDataSize = sizeof(uint32_t) + sizeof(char);


bool checkFlag(char flag) {
	return flag == GROUP_BY_MID_FLAG || flag == GROUP_BY_END_FLAG 
		|| flag == INTERNAL_PROTO_FLAG || flag == CONSUME_END_FLAG
		|| flag == CONSUME_MID_FLAG || flag == NETWORK_END_FLAG;
}

Network::Network(int argc, char** argv):
nei(CreateNodeEnvironment(argc, argv)),
networkThread(0) {
	CHECK(nei != 0, "Null node environment interface");
}

char* getMetaFlagPointer(char *data, int packetLength) {
	return data + (packetLength + metaFlagOffset);
}

const char* getRecipentPointer(const char* data, int packetLength) {
	return data + packetLength;
}

Network::~Network() {
	delete nei;
}

void networkThreadFun() {
	while (true) {
		Network::Packet packet;
		std::size_t len;
		packet.data = network->nei->ReadPacketBlocking(&len);
		CHECK(len >= metaDataSize, "Packet length too small - bad format");

		packet.length = len - metaDataSize;
		uint32_t recipentId = *(uint32_t*) getRecipentPointer(packet.data, packet.length);
		packet.flag = *getMetaFlagPointer(packet.data, packet.length);
		CHECK(checkFlag(packet.flag), "Invalid packet flag read");

		if (packet.flag == NETWORK_END_FLAG) {
			delete [] packet.data;
			break;
		}

		CHECK(recipentId == 0 || recipentId == 1, "Bad recipent id in incoming packet");

		{
			boost::unique_lock<boost::mutex> lock(network->bufferMutex);
			network->buffer[recipentId].push_back(packet);
			network->dataReadyForThreadCond[recipentId].notify_one();
		}
	}
}

void Network::startNetworkThread() {
	CHECK(network != 0, "Network not initialized");
	CHECK(networkThread == 0, "Network thread already running");
	
	networkThread = new boost::thread(networkThreadFun);
}
 
void Network::stopNetworkThread() {
	network->sendToThread(my_node_number(), 0, "", 0, NETWORK_END_FLAG);
	
	networkThread->join();
	delete networkThread;
	networkThread = 0;
}

uint32_t Network::getThreadId() const {
	return *threadId;
}

void Network::setThreadId(uint32_t threadId_) {
	CHECK(threadId_ == 0 || threadId_ == 1, "Thread id must be either 0 or 1");
	threadId.reset(new uint32_t(threadId_));
}

char* Network::read(std::size_t* data_len, char* flag) {
	return readAsThread(*threadId, data_len, flag);
}

void Network::send(uint32_t node, const char* data, std::size_t data_len, char flag) {
	sendToThread(node, 1 - *threadId, data, data_len, flag);
}

char* Network::readAsThread(uint32_t asThreadId, std::size_t* data_len, char* flag) {
	boost::unique_lock<boost::mutex> lock(bufferMutex);

	while (buffer[asThreadId].empty()) {
		dataReadyForThreadCond[asThreadId].wait(lock);
	}

	*data_len = buffer[asThreadId].front().length;
	char* data = buffer[asThreadId].front().data;
	*flag = buffer[asThreadId].front().flag;
	buffer[asThreadId].pop_front();

	return data;
}

void Network::sendToThread(uint32_t node, uint32_t toThreadId, const char* data, std::size_t data_len, char flag) {
	CHECK(checkFlag(flag), "Invalid packet flag sent");
	std::size_t sentDataLen = data_len + metaDataSize;
	boost::scoped_array<char> sentData(new char[sentDataLen]);

	memcpy(sentData.get(), data, data_len);
	*(uint32_t*) (getRecipentPointer(sentData.get(), data_len)) = toThreadId;
	*(getMetaFlagPointer(sentData.get(), data_len)) = flag;

	nei->SendPacket(node, sentData.get(), sentDataLen);
}

uint32_t Network::my_node_number() const {
	return nei->my_node_number();
}

uint32_t Network::nodes_count() const {
	return nei->nodes_count();
}

char* Network::readRaw(std::size_t* data_len) {
	return nei->ReadPacketBlocking(data_len);
}

void Network::sendRaw(uint32_t node, const char* data, std::size_t data_len) {
	return nei->SendPacket(node, data, data_len);
}

DataSourceInterface* Network::openDataSourceFile(int source_file_id) {
	return nei->OpenDataSourceFile(source_file_id);
}

DataSinkInterface* Network::openDataSink() {
	return nei->OpenDataSink();
}
